set hive.exec.dynamic.partition= true;
set hive.exec.dynamic.partition.mode= 'nonstrict';
set hive.exec.max.dynamic.partitions.pernode= 200;
set hive.exec.max.dynamic.partitions=200 ;
--网点信息维度表
with network_expand as(
 select
      code,                      --网点编码
      name,                      --网点名字
      agent_name as subordinate_agent_name,    --所属代理区
      agent_id as subordinate_agent_id,      --所属代理区ID
      agent_code as subordinate_agent_code,    --所属代理区code
      city_id,                   --城市ID
      city_desc city_desc                  --城市名字
 from jms_dim.dim_network_whole_massage
where is_delete=1
    and is_enable=1
    and name not like 'BEST%'
 ),
     waybill_incre as(
SELECT
     a.waybill_no,                 --运单号
     a.pick_network_code as input_network_code ,         --录入网点
     a.package_inside_charge_weight as settlement_weight ,--内部计费重量
     a.is_refund,             --运单状态 1:已退件2:已转寄3:未退件4:退转件
     a.sign_network_code,
     a.sign_network_name,
     a.destination_name,
     b.subordinate_agent_code,  --始发代理区
     b.subordinate_agent_name,  --始发代理区
     dt                         --订单录入日期
from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt a
left join network_expand b
on a.pick_network_code=b.code
where dt>date_add('{{ execution_date | cst_ds }}', -20)
and dt<=date_add('{{ execution_date | cst_ds }}', -6)

),
 --网点到件
 sitearrival_base as(
 select
    waybill_no                                   --运单号
     ,subordinate_agent_code   --代理区code
    ,subordinate_agent_name  --代理区
    ,org_code,                                     --网点code
    org_name,                                     --扫描网点
    weight,                                       --重量
    scan_time,                                    --时间
    scan_type,                                     --扫描类型
    volume_weight                                 --体积重
 from (
 select
     waybill_no,                                  --运单号
     round(length*width*height/8000,2) as volume_weight,   --网点体积重
     a.scan_site as org_name,                       --扫描网点
     a.scan_site_code as org_code,                  --网点code
     a.weight,                                      --重量
     a.scan_time,                                    --时间
     a.scan_type,                                     --扫描类型
     b.subordinate_agent_code   --代理区code
     ,b.subordinate_agent_name  --代理区
     ,row_number() over(partition by waybill_no,scan_site order by weight desc ) as rn
 from jms_dwd.dwd_tab_barscan_sitearrival_base_dt a
left join network_expand b
on a.scan_site_code=b.code
 where dt>date_add('{{ execution_date | cst_ds }}', -20)
 and dt<='{{ execution_date | cst_ds }}'
 and weight is not null
  and weight <> 0
 and waybill_no not like 'B%'
 )temp
  where temp.rn=1
 ),

 --中心到件表 中心表对运单中心去重
 centerarrival_bas_1 as(
 select
     waybill_no                                           --运单号
    ,subordinate_agent_code   --代理区code
    ,subordinate_agent_name  --代理区
     ,org_code                                             --中转中心code
     ,org_name                                           --中转中心名字
     ,weight                                             --中转中心称重
     ,scan_time                                            --扫描时间
     ,scan_type  --扫描类型
     ,center_package_volume as volume_weight                --中转中心体积重
    from (
     select
         waybill_no,                                       --运单号
         scan_site_code as  org_code,               --中转中心code
         scan_site as org_name,                     --中转中心名字
         weight,                                           --中转中心称重
         round(length*width*height/8000,2) as center_package_volume,--中转中心体积重
         scan_time,                                        --扫描时间
            scan_type,  --扫描类型
          b.subordinate_agent_code   --代理区code
        , b.subordinate_agent_name  --代理区
        , row_number() over(partition by waybill_no,scan_site_code order by weight desc  ) rn
     from jms_dwd.dwd_tab_barscan_centerarrival_base_dt a
left join network_expand b
on a.scan_site_code=b.code
     where dt>date_add('{{ execution_date | cst_ds }}', -20)
            and dt<='{{ execution_date | cst_ds }}'
 and weight is not null
 and weight <> 0
 and waybill_no not like 'B%'
 )temp
 where temp.rn=1
 ),
 --入仓表
 warehousing_base as (
          select waybill_no,                              --运单号
                 org_code,                                --网点code
                 org_name,                                --网点名字
                 weight,                                  --重量
                 scan_time,                                        --扫描时间
                 scan_type,   --扫描类型
                 subordinate_agent_code,   --代理区code
                 subordinate_agent_name,  --代理区
                 warehouse_volume_weight as volume_weight --体积重
          from (
                   select waybill_no,                                                                                                     --运单号
                          scan_site_code                                                                      as org_code,--网点code
                          --scan_site_id,  --网点ID
                          input_site                                                                          as org_name,                --网点名字
                          weight,                                                                                                         --重量
                          scan_time,                                        --扫描时间
                          round(length * width * height / 8000, 2)                                            as warehouse_volume_weight, --体积重
                          scan_type,   --扫描类型
                            b.subordinate_agent_code   --代理区code
                          , b.subordinate_agent_name  --代理区
                          ,row_number() over (partition by waybill_no,scan_site_code order by weight desc ) as rn
                   from jms_dwd.dwd_tab_barscan_warehousing_base_dt a
                   left join network_expand b
                       on a.scan_site_code=b.code
                   where dt>date_add('{{ execution_date | cst_ds }}', -20)
                     and dt<='{{ execution_date | cst_ds }}'
                     and weight is not null
                      and weight <> 0
                     and waybill_no not like 'B%'
               ) temp
          where temp.rn = 1
      ),
      --将入仓和揽收先进进行合并处理
         warehousing_union_taking as(
             select
                 waybill_no,                         --运单号
                 org_code,                           --网点code
                 max(org_name) as org_name,                           --网点名字
                 max(weight) as weight,                             --重量
                 min(scan_time)as scan_time,    --时间
                  max(scan_type) as scan_type,   -- 扫描类型
                  max(subordinate_agent_code) as subordinate_agent_code,
                  max(subordinate_agent_name) as subordinate_agent_name,
                 max(volume_weight)as volume_weight --揽收体积重
             from(
                  select
                     waybill_no,                         --运单号
                     org_code,                           --网点code
                     org_name,                           --网点名字
                     weight,                             --重量
                         scan_type,   -- 扫描类型
                     scan_time,    --时间
                      subordinate_agent_code,   --代理区code
                    subordinate_agent_name,  --代理区
                     volume_weight --揽收体积重
                  from warehousing_base
               )temp
             group by waybill_no,org_code

          ),
 --将网点和中心数据合并然后进行重量排序
 site_union_center as(
 select
    waybill_no,
    max(first_org_code) as  first_org_code,                                   --机构一编码
    max(first_org_name) as   first_org_name,                                  --机构一名字
    max(first_weight)    as first_weight,                                     --机构一重量
    max(first_volume_weight) as first_volume_weight,                          --机构一体积重
    max(first_scan_type) as     first_scan_type,                              --机构二扫描类型
    max(second_org_code) as second_org_code,                                  --机构二code
    max(second_org_name) as second_org_name,                                  --机构二名字
    max(second_weight) as second_weight,                                      --机构二重
    max(second_volume_weight) as second_volume_weight,                        --机构二体积重
    max(second_scan_type) as second_scan_type,                            --机构二扫描类型
    max(third_org_code) as third_org_code,                                    --机构三code
    max(third_org_name) as third_org_name ,                                   --机构三名字
    max(third_weight) as third_weight,                                        --机构三重量
    max(third_volume_weight) as third_volume_weight,                           --机构三体积重
    max(third_scan_type) as third_scan_type ,                                     --机构三扫描类型
    max(is_reweigh) as is_reweigh   --是否重复称重
 from (
    select
       temp.waybill_no,                                                     --运单号
       case when rn=1  then  org_code end as first_org_code,             --机构一编码
       case when rn=1  then  org_name end as first_org_name,             --机构一名字
       case when rn=1  then  weight end as first_weight,                 --机构一重量
       case when rn=1  then  volume_weight end as first_volume_weight,   --机构一体积重
       case when rn=1  then  scan_type end as first_scan_type,   --机构一扫描类型
       case when rn=2  then  org_code end as second_org_code,            --机构二code
       case when rn=2  then  org_name end as second_org_name,            --机构二名字
       case when rn=2  then  weight end as second_weight,                --机构二重
       case when rn=2  then  volume_weight end as second_volume_weight,  --机构二体积重
       case when rn=2  then  scan_type end as second_scan_type,  --机构二扫描类型
       case when rn=3  then  org_code end as third_org_code,             --机构三code
       case when rn=3  then  org_name end as third_org_name,             --机构三名字
       case when rn=3  then  weight end as third_weight,                 --机构三重量
       case when rn=3  then  scan_type end as third_scan_type,                 --机构三扫描类型
       case when rn=3  then  volume_weight end as third_volume_weight,    --机构三体积重
       if(temp.subordinate_agent_code=waybill_incre.subordinate_agent_code,0,1) as is_reweigh   --是否重复称重
   from (
       select
         waybill_no,              --运单号
         subordinate_agent_code,   --代理区code
        subordinate_agent_name,   --代理区code
         org_code,                --机构code
         org_name,                --扫描机构
         weight,                  --重量
         volume_weight,           --体积重
              scan_type, -- 扫描类型
         row_number() over(partition by waybill_no order by weight desc,scan_time asc) as rn
      from (
          select
           waybill_no,         --运单号
           subordinate_agent_code,   --代理区code
           subordinate_agent_name,   --代理区code
           org_code,           --机构code
           org_name,           --扫描机构
           weight,             --重量
           scan_time,          --时间
           scan_type, -- 扫描类型
           volume_weight       --体积重
         from centerarrival_bas_1
         UNION ALL
         select
           waybill_no,          --运单号
          subordinate_agent_code,   --代理区code
           subordinate_agent_name,   --代理区code
           org_code,            --机构code
           org_name,            --扫描机构
           weight,              --重量
           scan_time,          --时间
                scan_type, -- 扫描类型
           volume_weight        --体积重
         from sitearrival_base
         union all
         select
                waybill_no,   --运单号
                subordinate_agent_code,   --代理区code
                subordinate_agent_name,   --代理区code
                org_code,     --机构code
                org_name,     --扫描机构
                weight,       --重量
                scan_time,     --时间
                scan_type, -- 扫描类型
                volume_weight --体积重
         from warehousing_union_taking
      )tmp1
   )temp
     left join  waybill_incre
     on temp.waybill_no=waybill_incre.waybill_no
) temp1
group by waybill_no

 )

--插入语句
   insert overwrite table jms_dm.dm_daily_waybill_weight_difference_detail_new_dt partition(dt)
      select
         a.waybill_no,                                                      --运单号
         b.subordinate_agent_id,          --所属代理区ID
         b.subordinate_agent_code,        --所属代理区code
         b.subordinate_agent_name,        --所属代理区
         b.city_id,                       --城市ID
         b.city_desc,                     --城市名字
         b.code,                          --网点编码
         b.name,                          --网点名字
         a.settlement_weight,             --结算重量(内部结算重量)
         c.first_org_code,                --机构一编码
         c.first_org_name,                --机构一名字
         c.first_weight,                  --机构一重量
         c.first_volume_weight,           --机构一体积重
         c.second_org_code,               --机构二code
         c.second_org_name,               --机构二名字
         c.second_weight,                 --机构二重
         c.second_volume_weight,          --机构二体积重
         c.third_org_code,                --机构三code
         c.third_org_name,                --机构三名字
         c.third_weight,                  --机构三重量
         c.third_volume_weight,           --机构三体积重
         a.dt as input_date,              --运单录入日期
         a.is_refund,             --运单状态 1:已退件2:已转寄3:未退件4:退转件
         c.first_scan_type,          --机构1扫描类型
         c.second_scan_type,         --机构2扫描类型
         c.third_scan_type,          --机构3扫描类型
         c.is_reweigh,      --是否重复称重
         a.sign_network_code,    --签收网点code
         a.sign_network_name,   --签收网点
         a.destination_name,    --目的地
         dt
     from waybill_incre a           ---订单表
     join network_expand b          --网点信息维度表
       on a.input_network_code=b.code
     join site_union_center c       --网点和中心合并并排序表
       on a.waybill_no=c.waybill_no
       -- and a.settlement_weight+0.1 <c.first_weight
      distribute by dt, abs(hash(waybill_no)) % 60 ;
    -- 再次出明细
   insert overwrite table jms_dm.dm_daily_waybill_weight_difference_detail_dt partition(dt)
    select
             waybill_no,                                                      --运单号
             subordinate_agent_id,          --所属代理区ID
             subordinate_agent_code,        --所属代理区code
             subordinate_agent_name,        --所属代理区
             city_id,                       --城市ID
             city_desc,                     --城市名字
             code,                          --网点编码
             name,                          --网点名字
             settlement_weight,             --结算重量(内部结算重量)
             first_org_code,                --机构一编码
             first_org_name,                --机构一名字
             first_weight,                  --机构一重量
             first_volume_weight,           --机构一体积重
             second_org_code,               --机构二code
             second_org_name,               --机构二名字
             second_weight,                 --机构二重
             second_volume_weight,          --机构二体积重
             third_org_code,                --机构三code
             third_org_name,                --机构三名字
             third_weight,                  --机构三重量
             third_volume_weight,           --机构三体积重
             dt as input_date,              --运单录入日期
             is_refund,             --运单状态 1:已退件2:已转寄3:未退件4:退转件
             first_scan_type,          --机构1扫描类型
             second_scan_type,         --机构2扫描类型
             third_scan_type,          --机构3扫描类型
              is_reweigh,      --是否重复称重
              sign_network_code,    --签收网点code
              sign_network_name,   --签收网点
              destination_name,    --目的地
             dt
    from jms_dm.dm_daily_waybill_weight_difference_detail_new_dt
    where dt>date_add('{{ execution_date | cst_ds }}', -20)
    and dt<=date_add('{{ execution_date | cst_ds }}', -6)
    and settlement_weight+0.1 <first_weight
    distribute by dt,10;